package com.mininglamp.offline

import breeze.numerics.sqrt
import com.mongodb.spark.MongoSpark
import com.mongodb.spark.config.ReadConfig
import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
  * Project: ECommerceRecommendSystem
  * Package: com.mininglamp.offline
  * Description: ALS模型训练和参数选取.
  * 原理：通过计算每个商品评分推荐预测值和真实值之间的差值平方和均值取平方根，也就是得到 RMSE.选取RMSE值最小的一组作为参数最优解
  *
  * Created by ZhouPeng on 2021/12/29 11:29
  **/
object ALStrainer {

  val MONGODB_RATING_COLLECTION = "Rating"

  def main(args: Array[String]): Unit = {

    //基本配置
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://master:27017/recommender",
      "mongo.db" -> "recommender"
    )

    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("ALSTrainer")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    import sparkSession.implicits._
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))


    val readConfig = ReadConfig(Map(
      "uri" -> mongoConfig.mongoUri,
      "collection" -> MONGODB_RATING_COLLECTION
    ))

    val ratingRDD = MongoSpark.load(sparkSession.sparkContext, readConfig).map(
      document => {
        ProductRating(document.getInteger("userId"), document.getInteger("productId"),
          document.getDouble("score"), document.getLong("timestamp"))
      }
    ).map(productRating => Rating(productRating.userId, productRating.productId, productRating.score)).cache()

    //将数据集随机切分成训练集和测试集
    val splits = ratingRDD.randomSplit(Array(0.7, 0.3))
    val trainingRDD = splits(0)
    val testingRDD = splits(1)

    //todo 核心实现：输出最优参数
    adjustALSParams(trainingRDD, testingRDD)

    sparkSession.stop()
  }

  /**
    * 通过测试集和训练集输出最优参数解
    *
    * @param traingRDD  训练集
    * @param testingRDD 测试集
    */
  def adjustALSParams(traingRDD: RDD[Rating], testingRDD: RDD[Rating]): Unit = {

    val paramResult = for (rank <- Array(5, 10, 20, 50); lambda <- Array(1, 0.1, 0.01))
      yield {
        val model = ALS.train(traingRDD, rank, 10, lambda)
        //通过现有模型和测试数据集算出RMSE的值
        val rmse = getRMSE(model, testingRDD)
        (rank, lambda, rmse)
      }

    //println(paramResult.minBy(_._3))
    paramResult.foreach(println)
  }


  /**
    * 计算RMSE值
    *
    * @param model
    * @param testingRDD
    * @return
    */
  def getRMSE(model: MatrixFactorizationModel, testingRDD: RDD[Rating]): Double = {

    val userProduct = testingRDD.map(Rating => (Rating.user, Rating.product))
    //商品评分预测值
    val predictRating = model.predict(userProduct)

    //按照rmse公式，将真实评分值和预测评分制做((user,product),score)处理，做连接计算两者分值的差
    //真实评分值
    val realScore = testingRDD.map(Rating => ((Rating.user, Rating.product), Rating.rating))
    //预测评分值
    val predictScore = predictRating.map(Rating => ((Rating.user, Rating.product), Rating.rating))

    //求 rmse值
    sqrt(realScore.join(predictScore).map {
      case ((user, product), (real, predict)) => {
        val err = real - predict
        err * err
      }
    }.mean())
  }
}
